package com.corn.flink.lesson4;

import cn.hutool.core.collection.CollUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * source watermark operator demo
 *
 * @author JimWu
 * @date 2023/3/2 10:10
 **/
public class FlinkWatermarkSourceOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Event> ds = env.addSource(new RichSourceFunction<Event>() {
            private boolean flag = true;

            private String[] types = {"add", "delete", "update", "select"};


            @Override
            public void run(SourceContext<Event> ctx) throws Exception {
                while (flag) {
                    Event event = new Event(types[ThreadLocalRandom.current().nextInt(3)], System.currentTimeMillis());
                    // 采集数据并携带时间戳
                    ctx.collectWithTimestamp(event, event.timestamp);
                    // 发送水位线
                    ctx.emitWatermark(new Watermark(event.timestamp));
                    Thread.sleep(TimeUnit.SECONDS.toMillis(1));
                }
            }

            @Override
            public void cancel() {
                this.flag = false;
            }
        });

        ds.print();

        env.execute();
    }
}
